﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Lucene.Net.Index;
using Apache.Cassandra;
using Lucandra.Convenience;
using Lucandra.Net.Connection.Pooling;
using System.Diagnostics;
using Lucandra.Net.Connection;

namespace Lucandra.Net
{
    internal sealed class TermInfoReader
    {
        public int MaximumReadAhead { get; private set; }
        public Term CurrentTerm { get; private set; }
        public LucandraTermInfo CurrentTermInfo { get; private set; }
        public string IndexName { get; private set; }
        public string CurrentFieldName
        {
            get { return this.CurrentTermInfo == null ? null : this.CurrentTermInfo.FieldName; }
        }
        public int Position { get; private set; }
        public bool IsOpen
        {
            get { return this.TermInfoCache != null; }
        }
        public bool UseKeyspacePerIndex { get; private set; }
        public string Keyspace { get; private set; }

        private LucandraTermInfo[] TermInfoCache { get; set; }
        private Dictionary<Term, int> TermToTermInfoPositionMap { get; set; }

        public TermInfoReader(string indexName, int maximumReadAhead, bool useKeyspacePerIndex, string keyspace)
        {
            this.IndexName = indexName;
            this.MaximumReadAhead = maximumReadAhead;
            this.TermToTermInfoPositionMap = new Dictionary<Term, int>();
            this.Position = -1;
            this.CurrentTerm = null;
            this.CurrentTermInfo = null;
            this.UseKeyspacePerIndex = useKeyspacePerIndex;
            this.Keyspace = keyspace;
        }

        public bool Next()
        {
            if (this.TermInfoCache == null)
            {
                this.PositionAt(new Term(""));
                return (this.TermInfoCache.Length > 0);
            }
            else if (this.TermInfoCache.Length == 0)
            {
                return false;
            }
            else if (this.Position == (this.TermInfoCache.Length - 1))
            {
                this.PositionAfter(this.CurrentTerm);
                return (this.TermInfoCache.Length > 0);
            }
            else
            {
                this.Position++;
                this.CurrentTerm = this.TermInfoCache[this.Position].GetTerm();
                this.CurrentTermInfo = this.TermInfoCache[this.Position];
                return true;
            }
        }

        public void PositionAfter(Term term)
        {
            int existingPosition = -1;
            if (this.Position < this.TermInfoCache.Length - 1 && this.TermToTermInfoPositionMap.TryGetValue(term, out existingPosition))
            {
                var oldTermCache = this.TermInfoCache;
                this.TermInfoCache = new LucandraTermInfo[oldTermCache.Length - existingPosition];
                this.TermToTermInfoPositionMap.Clear();
                Array.Copy(oldTermCache, existingPosition, this.TermInfoCache, 0, this.TermInfoCache.Length);
                for (int i = 0; i < this.TermInfoCache.Length; i++)
                    this.TermToTermInfoPositionMap.Add(this.TermInfoCache[0].GetTerm(), i);
                this.Next();
            }
            else
            {
                //string startTermKey = string.Format("{0}{1}{2}{3}{4}", this.IndexName, Settings.Delimeter, term.Field(), Settings.Delimeter, term.Text());
                string startTermKey = LucandraTermInfo.CreateTermKey(this.IndexName, term.Field(), term.Text(), this.UseKeyspacePerIndex);
                //string endTermKey = string.Format("{0}{1}{2}{3}{4}", this.IndexName, Settings.Delimeter, term.Field(), Settings.Delimeter, Settings.FinalToken);
                string endTermKey = "";

                // Create a new column parent for accessing [TermInfo].
                var columnParent = CassandraHelper.CreateColumnParent(Settings.TermInfoColumnFamily);
                // Create the slice range.
                var sliceRange = CassandraHelper.CreateSliceRange(true, int.MaxValue);
                // Create the slice predicate.
                var slicePredicate = CassandraHelper.CreateSlicePredicate(sliceRange);
                // Create the key range.
                var keyRange = CassandraHelper.CreateKeyRange(
                    Encoding.UTF8.GetBytes(startTermKey),
                    Encoding.UTF8.GetBytes(endTermKey),
                    this.MaximumReadAhead + 1);

                List<KeySlice> slices;

            DoSearch:

                // Get a reference to the client.  If a search context exists then use its client, otherwise borrow one from the pool.
                CassandraClient client;
                if (LucandraSearchContext.Current != null)
                    client = LucandraSearchContext.Current.Client;
                else
                {
                    client = CassandraConnectionPool.Borrow();
                    client.InnerClient.set_keyspace(this.Keyspace);
                }

                // Perform the search against Cassandra.
                try
                {
                    //slices = client.InnerClient.get_range_slices(columnParent, slicePredicate, keyRange, ConsistencyLevel.ONE);
                    slices = client.GetRangeSlices(columnParent, slicePredicate, keyRange, ConsistencyLevel.ONE);
                }
                catch (CassandraClientInvalidatedException)
                {
                    if (LucandraSearchContext.Current != null)
                        LucandraSearchContext.Current.InvalidateClient();
                    goto DoSearch;
                }
                finally
                {
                    if (LucandraSearchContext.Current == null)
                        client.Dispose();
                }

                this.TermInfoCache = new LucandraTermInfo[slices.Count - 1];
                this.TermToTermInfoPositionMap.Clear();
                int position = 0;

                foreach (var keySlice in slices)
                {
                    // Parse a TermInfo object from the term key (IndexName/FieldName/TermText).
                    var termInfo = LucandraTermInfo.CreateFromTermKey(Encoding.UTF8.GetString(keySlice.Key));

                    if (termInfo.GetTerm().Equals(term))
                        continue;

                    // Iterate through the supercolumns [TermInfo => *].
                    foreach (var column in keySlice.Columns)
                    {
                        string columnName = Encoding.UTF8.GetString(column.Column.Name);
                        var termInfoDocument = LucandraTermInfo.LucandraTermInfoDocument.FromCassandraKeyValue(columnName, column.Column.Value);
                        termInfo.AddTermInfoDocument(termInfoDocument);
                    }

                    if (LucandraSearchContext.Current != null)
                        LucandraSearchContext.Current.CacheTermInfo(termInfo);

                    this.TermInfoCache[position] = termInfo;
                    this.TermToTermInfoPositionMap[termInfo.GetTerm()] = position;
                    position++;
                }
            }

            if (this.TermInfoCache.Length > 0)
            {
                this.CurrentTerm = this.TermInfoCache[0].GetTerm();
                this.CurrentTermInfo = this.TermInfoCache[0];
                this.Position = 0;
            }
            else
                this.Position = -1;
        }

        public void PositionAt(Term term)
        {
            int existingPosition = -1;
            if (this.TermToTermInfoPositionMap.TryGetValue(term, out existingPosition))
            {
                this.Position = existingPosition;
                this.CurrentTerm = this.TermInfoCache[this.Position].GetTerm();
                this.CurrentTermInfo = this.TermInfoCache[this.Position];
                return;
            }
            else
            {
                //string startTermKey = string.Format("{0}{1}{2}{3}{4}", this.IndexName, Settings.Delimeter, term.Field(), Settings.Delimeter, term.Text());
                string startTermKey = LucandraTermInfo.CreateTermKey(this.IndexName, term.Field(), term.Text(), this.UseKeyspacePerIndex);
                //string endTermKey = string.Format("{0}{1}{2}{3}{4}", this.IndexName, Settings.Delimeter, term.Field(), Settings.Delimeter, Settings.FinalToken);
                string endTermKey = "";

                // Create a new column parent for accessing [TermInfo].
                var columnParent = CassandraHelper.CreateColumnParent(Settings.TermInfoColumnFamily);
                // Create the slice range.
                var sliceRange = CassandraHelper.CreateSliceRange(true, int.MaxValue);
                // Create the slice predicate.
                var slicePredicate = CassandraHelper.CreateSlicePredicate(sliceRange);
                // Create the key range.
                var keyRange = CassandraHelper.CreateKeyRange(
                        Encoding.UTF8.GetBytes(startTermKey),
                        Encoding.UTF8.GetBytes(endTermKey),
                        this.MaximumReadAhead);

                List<KeySlice> slices;

            DoSearch:

                // Get a reference to the client.  If a search context exists then use its client, otherwise borrow one from the pool.
                CassandraClient client;
                if (LucandraSearchContext.Current != null)
                    client = LucandraSearchContext.Current.Client;
                else
                {
                    client = CassandraConnectionPool.Borrow();
                    client.InnerClient.set_keyspace(this.Keyspace);
                }

                // Perform the search against Cassandra.
            
                try
                {
                    //slices = client.InnerClient.get_range_slices(columnParent, slicePredicate, keyRange, ConsistencyLevel.ONE);
                    slices = client.GetRangeSlices(columnParent, slicePredicate, keyRange, ConsistencyLevel.ONE);
                }
                catch (CassandraClientInvalidatedException)
                {
                    if (LucandraSearchContext.Current != null)
                        LucandraSearchContext.Current.InvalidateClient();
                    goto DoSearch;
                }
                finally
                {
                    if (LucandraSearchContext.Current == null)
                        client.Dispose();
                }

                this.TermInfoCache = new LucandraTermInfo[slices.Count];
                this.TermToTermInfoPositionMap.Clear();
                int position = 0;

                foreach (var keySlice in slices)
                {
                    // Parse a TermInfo object from the term key (IndexName/FieldName/TermText).
                    var termInfo = LucandraTermInfo.CreateFromTermKey(Encoding.UTF8.GetString(keySlice.Key));

                    // Iterate through the columns [TermInfo => *].  These are all document pointers.
                    foreach (var column in keySlice.Columns)
                    {
                        string columnName = Encoding.UTF8.GetString(column.Column.Name);
                        var termInfoDocument = LucandraTermInfo.LucandraTermInfoDocument.FromCassandraKeyValue(columnName, column.Column.Value);
                        termInfo.AddTermInfoDocument(termInfoDocument);
                    }

                    if (LucandraSearchContext.Current != null)
                        LucandraSearchContext.Current.CacheTermInfo(termInfo);

                    this.TermInfoCache[position] = termInfo;
                    this.TermToTermInfoPositionMap[termInfo.GetTerm()] = position;
                    position++;
                }
            }

            if (this.TermInfoCache.Length > 0)
            {
                this.CurrentTerm = this.TermInfoCache[0].GetTerm();
                this.CurrentTermInfo = this.TermInfoCache[0];
                this.Position = 0;
            }
            else
                this.Position = -1;
        }
    }
}
